package com.gitee.melin.bee.core.jdbc.dialect.impl;

import com.aliyun.odps.Odps;
import com.aliyun.odps.Table;
import com.aliyun.odps.TableSchema;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.*;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.lang3.StringUtils;

public class MaxComputeJdbcDialect extends AbstractJdbcDialect {

    private Odps odps;

    public MaxComputeJdbcDialect(ConnectionInfo connectionInfo) {
        super((ConnectionInfo) null);
        this.connectionInfo = connectionInfo;

        Account account = new AliyunAccount(connectionInfo.getAccessKeyId(), connectionInfo.getSecretAccessKey());
        this.odps = new Odps(account);
        this.odps.setEndpoint(connectionInfo.getEndpoint());
    }

    @Override
    protected Driver getDriver() {
        return loadDriver(DataSourceType.MAXCOMPUTE.getDriverClass());
    }

    @Override
    public MetaDataSource testConnection(ConnectionInfo connectionInfo) {
        try {
            Account account = new AliyunAccount(connectionInfo.getAccessKeyId(), connectionInfo.getSecretAccessKey());
            Odps odps = new Odps(account);
            odps.setEndpoint(connectionInfo.getEndpoint());
            odps.tenant().reload();
            return new MetaDataSource();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        List<MetaTable> tableList = new ArrayList<>();
        odps.setDefaultProject(schemaName);
        for (Table table : odps.tables()) {
            MetaTable tableInfo = buildMetaTable(table);
            tableList.add(tableInfo);
        }
        return tableList;
    }

    @Override
    public List<SimpleMetaTable> getSimpleTables(String schemaName) {
        List<SimpleMetaTable> tableList = new ArrayList<>();
        odps.setDefaultProject(schemaName);
        for (Table table : odps.tables()) {
            SimpleMetaTable tableInfo = buildSimpleTable(table);
            tableList.add(tableInfo);
        }
        return tableList;
    }

    private static MetaTable buildMetaTable(Table table) {
        MetaTable tableInfo = new MetaTable();
        tableInfo.setTableName(table.getName());
        tableInfo.setComment(table.getComment());
        tableInfo.setLocation(table.getLocation());

        if (table.getType() == Table.TableType.MANAGED_TABLE) {
            tableInfo.setTableType(BeeTableType.TABLE);
        } else if (table.getType() == Table.TableType.EXTERNAL_TABLE) {
            tableInfo.setTableType(BeeTableType.EXTERNAL_TABLE);
        } else if (table.getType() == Table.TableType.VIRTUAL_VIEW) {
            tableInfo.setTableType(BeeTableType.VIEW);
        } else {
            tableInfo.setTableType(BeeTableType.MATERIALIZED_VIEW);
        }
        return tableInfo;
    }

    private static SimpleMetaTable buildSimpleTable(Table table) {
        SimpleMetaTable tableInfo = new SimpleMetaTable();
        tableInfo.setTableName(table.getName());
        tableInfo.setComment(table.getComment());

        if (table.getType() == Table.TableType.MANAGED_TABLE) {
            tableInfo.setTableType(BeeTableType.TABLE);
        } else if (table.getType() == Table.TableType.EXTERNAL_TABLE) {
            tableInfo.setTableType(BeeTableType.EXTERNAL_TABLE);
        } else if (table.getType() == Table.TableType.VIRTUAL_VIEW) {
            tableInfo.setTableType(BeeTableType.VIEW);
        } else {
            tableInfo.setTableType(BeeTableType.MATERIALIZED_VIEW);
        }
        return tableInfo;
    }

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.MAXCOMPUTE;
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName, boolean containColumn) {
        if (StringUtils.isBlank(schemaName)) {
            throw new IllegalStateException("schemaName can not blank");
        }

        if (StringUtils.isBlank(tableName)) {
            throw new IllegalStateException("tableName can not blank");
        }
        return this.getTable(schemaName, tableName);
    }

    protected MetaTable getTable(String schemaName, String tableName) {
        odps.setDefaultProject(schemaName);
        Table table = odps.tables().get(schemaName, tableName);
        MetaTable tableInfo = buildMetaTable(table);
        TableSchema tableSchema = table.getSchema();

        List<MetaColumn> columns = buildTableColumns(tableSchema);
        tableInfo.setColumns(columns);
        return tableInfo;
    }

    private static List<MetaColumn> buildTableColumns(TableSchema tableSchema) {
        List<MetaColumn> columns = new ArrayList<>();
        tableSchema.getColumns().forEach(column -> {
            MetaColumn field = new MetaColumn();
            field.setColumnName(column.getName());
            field.setDataType(column.getTypeInfo().getTypeName());
            field.setComment(column.getComment());
            field.setNullable(column.isNullable());
            columns.add(field);
        });

        tableSchema.getPartitionColumns().forEach(column -> {
            MetaColumn field = new MetaColumn();
            field.setColumnName(column.getName());
            field.setDataType(column.getTypeInfo().getTypeName());
            field.setComment(column.getComment());
            field.setPartitionField(true);
            columns.add(field);
        });
        return columns;
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName) {
        odps.setDefaultProject(schemaName);
        Table table = odps.tables().get(schemaName, tableName);
        TableSchema tableSchema = table.getSchema();
        return buildTableColumns(tableSchema);
    }

    @Override
    protected String getDDLColumnName(BeeTableType type) {
        throw new UnsupportedOperationException("not support");
    }
}
